package com.mashibing.service.impl.zookeeperImpl;

import com.mashibing.internalcommon.constant.CommonStatusEnum;
import com.mashibing.internalcommon.dto.ResponseResult;
import com.mashibing.internalcommon.request.DriverGrabRequest;
import com.mashibing.service.GrabService;
import com.mashibing.service.OrderService;
import org.apache.commons.lang.StringUtils;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

@Service("grabByZookeeperService")
public class GrabByZookeeperServiceImpl implements GrabService {

    @Autowired
    private OrderService orderService;

    @Value("${zookeeper.address}")
    private String address;

    @Value("${zookeeper.timeout}")
    private int timeout;

    @Override
    public ResponseResult grab(DriverGrabRequest driverGrabRequest) {

        String orderIdKey = driverGrabRequest.getOrderId() + "";
        ResponseResult<Void> responseResult = null;

        try {
            // 获取锁
            CountDownLatch countDownLatch = new CountDownLatch(1);
            ZooKeeper zooKeeper = new ZooKeeper(address, timeout, watchedEvent -> {
                if (Watcher.Event.KeeperState.SyncConnected == watchedEvent.getState()) {
                    countDownLatch.countDown();
                }
            });
            countDownLatch.await();

            // 生成永久节点
            String parentNode = "/order-" + orderIdKey;
            Stat exists = zooKeeper.exists(parentNode, false);
            if (exists == null){
                zooKeeper.create(parentNode,"".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);
            }

            // 生成临时子节点
            String seq = "/seq";
            String s = zooKeeper.create(parentNode + seq, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            boolean ifLock = false;
            if (StringUtils.isNotBlank(s)) {
                List<String> children = zooKeeper.getChildren(parentNode, false);
                Collections.sort(children);
                if (!children.isEmpty() && (parentNode + "/" + children.get(0)).equals(s)) {
                    ifLock = true;
                    // 执行业务代码
                    System.out.println("开始锁 Zookeeper div");
                    responseResult = orderService.driverGrab(driverGrabRequest);
                    System.out.println("结束锁 Zookeeper div");
                }
            }
            if (!ifLock){
                responseResult = ResponseResult.fail(CommonStatusEnum.ORDER_CAN_NOT_GRAB.getCode(),CommonStatusEnum.ORDER_CAN_NOT_GRAB.getMessage(),null);
            }

            // 关闭连接
            zooKeeper.close();

        } catch (IOException | InterruptedException | KeeperException e) {
            throw new RuntimeException(e);
        }

        return responseResult;
    }


}
